package belf.migrate.engine.job.verify;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DatabaseNotExistException;
import belf.migrate.api.exception.TableNotExistException;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.DbObjectType;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.type.VerifyTableType;
import belf.migrate.api.util.PostLog;
import belf.migrate.core.jdbc.DBTypeMapping;
import belf.migrate.core.jdbc.TypeModel;
import belf.migrate.core.number.NumberFormatUtil;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.model.verify.*;
import belf.migrate.engine.job.verify.utils.ColumnTypeUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.math.BigDecimal;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.*;
import java.util.stream.Collectors;

import static belf.migrate.api.job.JobCode.EXECUTING;
import static belf.migrate.api.job.JobCode.FAILTURE;

@Slf4j
public class VerifyTableTool {

    private Job job;

    private JobContext jobContext;

    private AbstractJdbcCatalog sourceCatalog;

    private AbstractSinkCatalog sinkCatalog;

    public VerifyTableTool(JobContext jobContext, Job job) {
        this.jobContext = jobContext;
        this.sourceCatalog = jobContext.getSourceCatalog();
        this.sinkCatalog = jobContext.getSinkCatalog();
        this.job = job;
    }


    public VerifyTableResult run() {
        log.info("开始进行表对象校验!");
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "开始进行表对象校验!",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        VerifyTableResult tableResult = new VerifyTableResult();

        Set<String> tableVerifyStatue = new HashSet<>();

        //获取数据库信息
        String sourceDatabaseName = job.getTask().getTaskConf().getSourceConf().getDatabaseName();
        String sinkDatabaseName = job.getTask().getTaskConf().getSinkConf().getDatabaseName();
        String sourceSchemaName = job.getTask().getTaskConf().getSourceConf().getSchemaName();
        String sinkSchemaName = job.getTask().getTaskConf().getSinkConf().getSchemaName();

        QualifiersModel sourceQualifiersModel = jobContext.getSourceQualifiersModel();
        QualifiersModel sinkQualifiersModel = jobContext.getSinkQualifiersModel();

        try {
            List<String> sourceTables;
            List<String> sinkTables;
            if (CollectionUtils.isEmpty(job.getMissions())) {
                sourceTables = sourceCatalog.listTables(sourceDatabaseName, sourceSchemaName);
                sinkTables = sinkCatalog.listTables(sinkDatabaseName, sinkSchemaName);
            } else {
                sourceTables = job.getMissions();
                sinkTables = job.getMissions();
            }
            tableResult.setSourceObjectCount(sourceTables.size());
            tableResult.setSinkObjectCount(sinkTables.size());
            if (sourceTables.size() != sinkTables.size()) {
                tableResult.setSuccess(false);
                tableVerifyStatue.add(VerifyTableType.TABLE_SIZE_ERROR.getType());
            }
            List<VerifyTableModel> verifyTableModels = new ArrayList<>();

            int successCount = 0;
            //遍历所有表
            for (int index=0; index < sourceTables.size(); index++) {
                jobContext.addProgress(DbObjectType.TABLE.getTypeName(), sourceTables.size());
                Set<String> verifyErrorMsg = new HashSet<>();
                long startTime = System.currentTimeMillis();
                VerifyTableModel verifyTableModel = new VerifyTableModel();
                TablePath sourceTablePath = TablePath.of(sourceDatabaseName, sourceSchemaName, sourceTables.get(index), sourceQualifiersModel);
                //获取源表所有字段
                List<ColumnMetaData> sourceColumn = sourceCatalog.getAllColumnsBySql(sourceTablePath);

                String sinkTableName = sinkCatalog.caseSensitive() ? sourceTables.get(index).toUpperCase() : sourceTables.get(index);
                TablePath sinkTablePath = TablePath.of(sinkDatabaseName, sinkSchemaName, sinkTableName, sinkQualifiersModel);

                if (!sinkCatalog.tableExists(sinkTablePath)) {
                    String sourceRowCount = sourceCatalog.countDataForTable(sourceTablePath);
                    verifyTableModel.setSourceName(sourceTables.get(index));
                    verifyTableModel.setSourceColumnCount(sourceColumn.size());
                    verifyTableModel.setSourceRowCount(sourceRowCount);
                    tableResult.setSuccess(false);
                    verifyTableModel.setSuccess(false);
                    verifyErrorMsg.add(VerifyTableType.TABLE_NOT_EXIST.getDesc());
                    tableVerifyStatue.add(VerifyTableType.TABLE_NOT_EXIST.getType());
                } else {
                    //获取目标表的所有字段
                    List<ColumnMetaData> sinkColumn = sinkCatalog.getAllColumnsBySql(sinkTablePath);
                    String sourceRowCount = sourceCatalog.countDataForTable(sourceTablePath);
                    String sinkRowCount = sinkCatalog.countDataForTable(sinkTablePath);
                    if (sourceRowCount.equals(sinkRowCount)) {
                        //验证数据
                        VerifyRowResult rowResult = verifyDataWithCol(sourceCatalog, sinkCatalog, sourceTablePath, sinkTablePath, Integer.parseInt(sourceRowCount), job.getTask().getTaskConf().getTaskType());
                        verifyTableModel.setRowResult(rowResult);
                        verifyTableModel.setSourceName(sourceTables.get(index));
                        verifyTableModel.setSinkName(sinkTableName);
                        verifyTableModel.setSourceColumnCount(sourceColumn.size());
                        verifyTableModel.setSinkColumnCount(sinkColumn.size());
                        verifyTableModel.setSourceRowCount(sourceRowCount);
                        verifyTableModel.setSinkRowCount(sinkRowCount);
                        if (!rowResult.getStatue().isEmpty()) {
                            for (String s : rowResult.getStatue()) {
                                verifyErrorMsg.add(VerifyTableType.getByType(s).getDesc());
                            }
                            tableResult.setSuccess(false);
                            verifyTableModel.setSuccess(false);
                        }
                    } else {
                        verifyTableModel.setSourceName(sourceTables.get(index));
                        verifyTableModel.setSinkName(sinkTableName);
                        verifyTableModel.setSourceColumnCount(sourceColumn.size());
                        verifyTableModel.setSinkColumnCount(sinkColumn.size());
                        verifyTableModel.setSourceRowCount(sourceRowCount);
                        verifyTableModel.setSinkRowCount(sinkRowCount);
                        tableResult.setSuccess(false);
                        verifyTableModel.setSuccess(false);
                        verifyErrorMsg.add(VerifyTableType.ROW_SIZE_ERROR.getDesc());
                    }
                }
                verifyTableModel.setVerifyErrorMsg(verifyErrorMsg);
                //计算耗时
                long endTime = System.currentTimeMillis();
                String timeConsuming = String.valueOf((double) (endTime - startTime) / 1000);
                verifyTableModel.setTimeConsuming(String.format("%s秒", timeConsuming));
                verifyTableModels.add(verifyTableModel);
                if (verifyTableModel.isSuccess()) {
                    successCount++;
                }

                String curLogMsg = String.format("表 %s 校验已完成进度：%s / %s.", sourceTables.get(index), index + 1, sourceTables.size());
                PostLog.sendLogModel(
                        job.getJobId(),
                        job.getTask().getTaskType().getName(),
                        curLogMsg,
                        LogLevelType.INFO.getType(),
                        EXECUTING,
                        jobContext.getProgress());
                log.info(curLogMsg);
            }
            tableResult.setAllSize(sourceTables.size());
            tableResult.setSuccessCount(successCount);
            tableResult.setFailCount(sourceTables.size() - successCount);
            tableResult.setVerifyErrorMsg(tableVerifyStatue);
            tableResult.setObjectList(verifyTableModels);
            float successPercent = sourceTables.isEmpty() ? 0 : (float) successCount / sourceTables.size();
            float failPercent = sourceTables.isEmpty() ? 0 : (float) (sourceTables.size() - successCount) / sourceTables.size();
            tableResult.setSuccessPercent(NumberFormatUtil.numberToString(successPercent, "0.0%"));
            tableResult.setFailPercent(NumberFormatUtil.numberToString(failPercent, "0.0%"));

        } catch (CatalogException | DatabaseNotExistException | TableNotExistException e) {
            log.error("表校验失败！", e);
            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    String.format("表校验失败: %s", e.getMessage()),
                    LogLevelType.ERROR.getType(),
                    EXECUTING,
                    jobContext.getProgress());
            tableResult.setCode(FAILTURE);
        } finally {
            try {
                sinkCatalog.close();
                sourceCatalog.close();
            } catch (CatalogException e) {
                log.error("关闭数据源失败");
            }
        }
        log.info("表校验成功完成!");
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "表校验成功完成!",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        return tableResult;
    }

    /**
     * 验证数据
     * @param sourceCatalog
     * @param sinkCatalog
     * @param sourceTablePath
     * @param sinkTablePath
     * @param rowSize
     * @param taskType
     * @return
     */
    public VerifyRowResult verifyData(AbstractJdbcCatalog sourceCatalog, AbstractJdbcCatalog sinkCatalog, TablePath sourceTablePath, TablePath sinkTablePath, int rowSize, String taskType) {

        VerifyRowResult result = new VerifyRowResult();
        List<List<VerifyColumnModel>> columnModelList = new ArrayList<>();
        Set<String> statueList = new HashSet<>();
        //获取源库和目标库的字段映射表
        Map<String, TypeModel> mapping = DBTypeMapping.getInstance().getMapping(
                taskType);
        try {
            CatalogTable sourceCatalogTable = sourceCatalog.getTable(sourceTablePath);

            TableSchema tableSchema = sourceCatalogTable.getTableSchema();
            List<String> pkColumns = tableSchema.getPrimaryKey().getColumnNames();

            List<List<ColumnData>> sourceData;
            List<List<ColumnData>> sinkData;
            if (!pkColumns.isEmpty()) {
                String pk = pkColumns.get(0);
                List<String> pkList = sourceCatalog.getRandomPks(sourceTablePath, pk, 10, rowSize >= 10 ? rowSize - 10 : 0);
                sourceData = sourceCatalog.getRandomDataFromTable(sourceTablePath, pk, pkList);
                sinkData = sinkCatalog.getRandomDataFromTable(sinkTablePath, pk, pkList);
            } else {
                sourceData = sourceCatalog.getDataWithNoPk(sourceTablePath, 10, rowSize >= 10 ? rowSize - 10 : 0);
                sinkData = sinkCatalog.getDataWithNoPk(sinkTablePath, 10, rowSize >= 10 ? rowSize - 10 : 0);
            }

            if (sourceData.size() != sinkData.size()) {
                statueList.add(VerifyTableType.ROW_SIZE_ERROR.getType());
            }
            for (int i = 0; i < sourceData.size(); i++) {
                List<VerifyColumnModel> columnModels = new ArrayList<>();
                List<ColumnData> sourceRow = sourceData.get(i);
                List<ColumnData> sinkRow = sinkData.get(i);
                Map<String, ColumnData> sourceDataMap = sourceRow.stream().collect(
                        Collectors.toMap(ColumnMetaData -> ColumnMetaData.getColumnName().toLowerCase(Locale.ROOT),
                                ColumnData -> ColumnData));
                Map<String, ColumnData> sinkDataMap = sinkRow.stream().collect(
                        Collectors.toMap(ColumnMetaData -> ColumnMetaData.getColumnName().toLowerCase(Locale.ROOT),
                                ColumnData -> ColumnData));
                boolean isRowSuccess = true;
                for (String key : sourceDataMap.keySet()) {
                    boolean isColSuccess = true;
                    VerifyTableType columnType = VerifyTableType.OK;
                    if (!sinkDataMap.containsKey(key)) {
                        statueList.add(VerifyTableType.COLUMN_MATCH_ERROR.getType());
                        columnType = VerifyTableType.COLUMN_MATCH_ERROR;
                        isRowSuccess = false;
                        isColSuccess = false;
                    } else if (!sinkDataMap.get(key).getColumnTypeName().equalsIgnoreCase(mapping.get(sourceDataMap.get(key).getColumnTypeName().toUpperCase(Locale.ROOT)).getTo())) {
                        statueList.add(VerifyTableType.COLUMN_TYPE_MATCH_ERROR.getType());
                        columnType = VerifyTableType.COLUMN_TYPE_MATCH_ERROR;
                        isRowSuccess = false;
                        isColSuccess = false;
                    } else if (!sourceDataMap.get(key).getColumnValue().equals(sinkDataMap.get(key).getColumnValue())) {
                        statueList.add(VerifyTableType.DATA_MATCH_ERROR.getType());
                        columnType = VerifyTableType.DATA_MATCH_ERROR;
                        isRowSuccess = false;
                        isColSuccess = false;
                    }
                    if (!isRowSuccess || !isColSuccess) {
                        VerifyColumnModel columnModel = new VerifyColumnModel();
                        columnModel.setSourceColName(sourceDataMap.get(key).getColumnName());
                        columnModel.setSinkColName(sinkDataMap.get(key).getColumnName());
                        columnModel.setSourceColType(sourceDataMap.get(key).getColumnTypeName());
                        columnModel.setSinkColType(sinkDataMap.get(key).getColumnTypeName());
//                        columnModel.setSourceColValue(sourceDataMap.get(key).getColumnValue());
//                        columnModel.setSinkColValue(sinkDataMap.get(key).getColumnValue());
                        columnModel.setStatue(columnType.getType());
                        columnModel.setSuccess(isColSuccess);
                        columnModels.add(columnModel);
                    }
                }
                columnModelList.add(columnModels);
            }
//            result.setRows(columnModelList);
            result.setStatue(statueList);


        } catch (TableNotExistException | CatalogException e) {
            log.error("表校验:数据校验失败！", e);
        }
        return result;
    }

    public VerifyRowResult verifyDataWithCol(AbstractJdbcCatalog sourceCatalog, AbstractJdbcCatalog sinkCatalog, TablePath sourceTablePath, TablePath sinkTablePath, int rowSize, String taskType) {

        VerifyRowResult result = new VerifyRowResult();
//        List<List<VerifyColumnModel>> columnModelList = new ArrayList<>();
        List<VerifyColResult> colResultList = new ArrayList<>();
        Set<String> statueList = new HashSet<>();
        //获取源库和目标库的字段映射表
        Map<String, TypeModel> mapping = DBTypeMapping.getInstance().getMapping(
                taskType);
        try {
            CatalogTable sourceCatalogTable = sourceCatalog.getTable(sourceTablePath);

            TableSchema tableSchema = sourceCatalogTable.getTableSchema();
            List<String> pkColumns = tableSchema.getPrimaryKey().getColumnNames();

            List<List<ColumnData>> sourceData;
            List<List<ColumnData>> sinkData;
            if (!pkColumns.isEmpty()) {
                String pk = pkColumns.get(0);
                List<String> pkList = sourceCatalog.getRandomPks(sourceTablePath, pk, 10, rowSize >= 10 ? rowSize - 10 : 0);
                try {
                    sourceData = sourceCatalog.getRandomDataFromTable(sourceTablePath, pk, pkList);
                    sinkData = sinkCatalog.getRandomDataFromTable(sinkTablePath, pk, pkList);
                } catch (Exception e) {
                    log.error(String.format("表校验:%s 获取数据失败！", sourceTablePath.getFullName()), e);
                    result.setStatue(new HashSet<>(Collections.singletonList(VerifyTableType.TABLE_DATA_ERROR.getType())));
                    return result;
                }
            } else {
                sourceData = sourceCatalog.getDataWithNoPk(sourceTablePath, 10, rowSize >= 10 ? rowSize - 10 : 0);
                sinkData = sinkCatalog.getDataWithNoPk(sinkTablePath, 10, rowSize >= 10 ? rowSize - 10 : 0);
            }

            if (sourceData.size() != sinkData.size()) {
                statueList.add(VerifyTableType.ROW_SIZE_ERROR.getType());
            }

            //存储列信息
            List<ColumnMetaData> sourceColumns = sourceCatalog.getAllColumnsBySql(sourceTablePath);
            List<ColumnMetaData> sinkColumns = sinkCatalog.getAllColumnsBySql(sinkTablePath);
            for (int i = 0; i < sourceColumns.size(); i++) {
                Map<String, ColumnMetaData> sourceDataMap = sourceColumns.stream().collect(
                        Collectors.toMap(ColumnMetaData -> ColumnMetaData.getColumnName().toLowerCase(Locale.ROOT),
                                ColumnData -> ColumnData));
                Map<String, ColumnMetaData> sinkDataMap = sinkColumns.stream().collect(
                        Collectors.toMap(ColumnMetaData -> ColumnMetaData.getColumnName().toLowerCase(Locale.ROOT),
                                ColumnData -> ColumnData));
                ColumnMetaData sourceColumn = sourceDataMap.get(sourceColumns.get(i).getColumnName().toLowerCase(Locale.ROOT));
                String sourceColumnNameLower = sourceColumn.getColumnName().toLowerCase(Locale.ROOT);
                ColumnMetaData sinkColumn = sinkDataMap.get(sourceColumnNameLower);
                VerifyColResult colResult = new VerifyColResult();
                Set<String> statue = new HashSet<>();
                if (sinkColumn != null) {
                    colResult.setSinkColName(sinkDataMap.get(sourceColumnNameLower).getColumnName());
                    colResult.setSinkColType(sinkDataMap.get(sourceColumnNameLower).getColumnTypeName());
                    colResult.setSinkColSize(String.valueOf(sinkDataMap.get(sourceColumnNameLower).getColumnDisplaySize()));
                    if (!columnEqual(sourceColumn.getColumnTypeName(), sinkColumn.getColumnTypeName(), sourceCatalog, sinkCatalog, mapping)) {
                        statue.add(VerifyTableType.COLUMN_TYPE_MATCH_ERROR.getType());
                    }
                } else {
                    statue.add(VerifyTableType.COLUMN_MATCH_ERROR.getType());
                }
                colResult.setSourceComment(sourceColumns.get(i).getColumnComment());
                colResult.setSourceColSize(String.valueOf(sourceColumns.get(i).getColumnDisplaySize()));
                colResult.setSourceColType(sourceColumns.get(i).getColumnTypeName());
                colResult.setSourceColName(sourceColumns.get(i).getColumnName());
                colResult.setStatue(statue);
                if (!statue.isEmpty()) {
                    colResult.setSuccess(false);
                }
                statueList.addAll(statue);
                colResultList.add(colResult);
            }

            Map<String, VerifyColResult> colResultMap = colResultList.stream().collect(
                    Collectors.toMap(VerifyColResult -> VerifyColResult.getSourceColName().toLowerCase(Locale.ROOT),
                            VerifyColResult -> VerifyColResult));

            if (!sourceData.isEmpty() && sinkData.isEmpty()) {
                statueList.add(VerifyTableType.DATA_MATCH_ERROR.getType());
                for (VerifyColResult colResult : colResultList) {
                    colResult.addStatue(VerifyTableType.DATA_MATCH_ERROR.getType());
                    colResult.setDataEqual(false);
                    colResult.setSuccess(false);
                }
            } else if (sourceData.isEmpty() && sinkData.isEmpty()) {
                for (VerifyColResult colResult : colResultList) {
                    colResult.setDataEqual(true);
                }
            } else if (sourceData.size() != sinkData.size()) {
                statueList.add(VerifyTableType.ROW_SIZE_ERROR.getType());
            } else {
                for (int i = 0; i < sourceData.size(); i++) {
                    List<ColumnData> sourceRow = sourceData.get(i);
                    List<ColumnData> sinkRow = sinkData.get(i);
                    Map<String, ColumnData> sourceDataMap = sourceRow.stream().collect(
                            Collectors.toMap(ColumnData -> ColumnData.getColumnName().toLowerCase(Locale.ROOT),
                                    ColumnData -> ColumnData));
                    Map<String, ColumnData> sinkDataMap = sinkRow.stream().collect(
                            Collectors.toMap(ColumnData -> ColumnData.getColumnName().toLowerCase(Locale.ROOT),
                                    ColumnData -> ColumnData));
                    for (String key : sourceDataMap.keySet()) {
                        ColumnData sourceColumn = sourceDataMap.get(key);
                        ColumnData sinkColumn = sinkDataMap.get(key);
                        boolean equal = (sourceColumn.getColumnValue() != null && sinkColumn.getColumnValue() == null) ||
                                (sourceColumn.getColumnValue() == null && sinkColumn.getColumnValue() != null) ||
                                (sourceColumn.getColumnValue() != null && !equal(sourceColumn, sinkColumn, sourceCatalog, sinkCatalog));

                        if (equal) {
                            statueList.add(VerifyTableType.DATA_MATCH_ERROR.getType());
                            colResultMap.get(key).addStatue(VerifyTableType.DATA_MATCH_ERROR.getType());
                            colResultMap.get(key).setDataEqual(false);
                            colResultMap.get(key).setSuccess(false);
                        }
                    }
                }
                colResultList = colResultMap.values().stream().filter(VerifyColResult -> !VerifyColResult.isSuccess()).collect(Collectors.toList());
            }

            result.setCols(colResultList);

            result.setStatue(statueList);

        } catch (TableNotExistException | CatalogException e) {
            log.error("表校验:数据校验失败！", e);
        }
        return result;
    }

    public boolean columnEqual(String sourceColType, String sinkColType, AbstractJdbcCatalog sourceCatalog, AbstractJdbcCatalog sinkCatalog, Map<String, TypeModel> mapping) {
        sourceColType = sourceColType.replace("\"", "");
        sinkColType = sinkColType.replace("\"", "");
        return ColumnTypeUtil.columnTypeEqual(sourceColType, sinkColType, sourceCatalog.name(), sinkCatalog.name(), mapping);
    }

    public boolean equal(ColumnData sourceColumn, ColumnData sinkColumn, AbstractJdbcCatalog sourceCatalog, AbstractJdbcCatalog sinkCatalog) {
        Object source = sourceColumn.getColumnValue();
        Object sink = sinkColumn.getColumnValue();
        if (sourceCatalog.specialColumns().contains(sourceColumn.getColumnTypeName().toUpperCase(Locale.ROOT))) {
            return true;
        }
        try {
            if (source instanceof Timestamp) {
                return ((Timestamp) source).equals((Timestamp) sink);
            } else if (source instanceof Date) {
                return ((Date) source).equals((Date) sink);
            } else if (source instanceof Time) {
                return ((Time) source).equals((Time) sink);
            } else if (source instanceof BigDecimal) {
                return ((BigDecimal) source).compareTo((BigDecimal) sink) == 0;
            } else {
                return sinkCatalog.dataEqualByType(sinkColumn.getColumnTypeName(), source, sink);
            }
        } catch (Exception e) {
            return sinkCatalog.dataEqualByType(sinkColumn.getColumnTypeName(), source, sink);
        }
    }

}
